package ccnl.demo.util;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.*;
import org.apache.spark.sql.types.StructType;

import java.io.Serializable;
import java.util.ArrayList;

/**
 * Created by wong on 16/5/20.
 */
public class Once implements Serializable {
    public SQLContext sqlContext;
    public Once(SQLContext sqlContext) {
        this.sqlContext = sqlContext;
    }


    public void combineURLandAdID2(String path) {
        DataFrame df = sqlContext.read().parquet(path);
        StructType st = df.schema();
        JavaRDD<Row> jrdd = df.toJavaRDD();
        int len = df.columns().length;

        JavaRDD<Row> processed = jrdd.map(r -> {
            int index = r.fieldIndex("C0");
            if (r.getString(index).equals("108804")) {
                ArrayList al = new ArrayList();
                for (int i = 0; i < len; i ++) {
                    if (i == index) {
                        al.add("107401");
                    }else {
                        al.add(r.get(i));
                    }
                }
                return RowFactory.create(al.toArray());
            }else {
                return r;
            }
        });
        DataFrame newdf = sqlContext.createDataFrame(processed, st);
        newdf.write().mode(SaveMode.Overwrite).parquet("/user/hg/iqiyi5/whole_clean_combineurl_6.parquet");

    }

    public void combineURLandAdID(String path) {
        DataFrame df = sqlContext.read().parquet(path);
        StructType st = df.schema();
        JavaRDD<Row> jrdd = df.toJavaRDD();
        int len = df.columns().length;
        jrdd = helper(jrdd, len, "108396", "108390");
//        jrdd = helper(jrdd, len, "108391", "108395");
//        jrdd = helper(jrdd, len, "108909", "108922");
//
//        jrdd = helper(jrdd, len, "108967", "108455");
//        jrdd = helper(jrdd, len, "108803", "108455");
//        jrdd = helper(jrdd, len, "108804", "107401");

        DataFrame newdf = sqlContext.createDataFrame(jrdd, st);
        newdf.write().mode(SaveMode.Overwrite).parquet("/user/hg/iqiyi5/whole_clean_combineurl.parquet");
    }
    public JavaRDD<Row> helper(JavaRDD<Row> jrdd, int len, String before, String after) {
        JavaRDD<Row> processed = jrdd.map(r -> {
            int index = r.fieldIndex("C0");
            if (r.get(index).equals(before)) {
                ArrayList al = new ArrayList();
                for (int i = 0; i < len; i ++) {
                    if (i == index) {
                        al.add(after);
                    }else {
                        al.add(r.get(i));
                    }
                }
                return RowFactory.create(al.toArray());
            }else {
                return r;
            }
        });
        return processed;
    }
}
